package com.harboursoftware.xstorage.web;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import javax.servlet.ServletException;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.fs.FileSystem;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.ListMultimap;
import com.harboursoftware.xstorage.Constants;
import com.harboursoftware.xstorage.XStorageException;
import com.harboursoftware.xstorage.XStorageRuntimeException;
import com.harboursoftware.xstorage.ac.AccessControlPolicy;
import com.harboursoftware.xstorage.db.BucketDao;
import com.harboursoftware.xstorage.db.DaoManager;
import com.harboursoftware.xstorage.fs.HDFSFileSystem;
import com.harboursoftware.xstorage.fs.XFileSystem;
import com.harboursoftware.xstorage.model.XBucket;
import com.harboursoftware.xstorage.model.XElement;
import com.harboursoftware.xstorage.model.XObject;
import com.harboursoftware.xstorage.model.XService;
import com.harboursoftware.xstorage.model.XUser;
import com.harboursoftware.xstorage.operations.Operation;
import com.harboursoftware.xstorage.operations.OperationFactory;
import com.harboursoftware.xstorage.results.ErrorResult;
import com.harboursoftware.xstorage.results.Result;
import com.harboursoftware.xstorage.util.Configuration;

/**
 * <b>该类为线程安全</b>
 * 
 * @author Simon Leung
 * @since 0.1
 */
@SuppressWarnings("serial")
public class StorageServlet extends HttpServlet {
    private final static Logger log = LoggerFactory.getLogger(StorageServlet.class);
    // private final static String[] EMPTY_STRING_ARRAY = new String[0];
    private final static XService service = new XService();

    private static OperationFactory operationFatory;

    static {
        XFileSystem fileSystem = initFileSystem();
        operationFatory = new OperationFactory(fileSystem);
    }

    private static XFileSystem initFileSystem() {
        // TODO:重连机制，如果连接在中途断开后，是否要重新连接
        // TODO:不应该在这里初始化XFileSystem
        org.apache.hadoop.conf.Configuration hadoopConfig = new org.apache.hadoop.conf.Configuration();
        Configuration xstorageConfig = Configuration.getInstance();
        hadoopConfig.set("fs.default.name", xstorageConfig.getString("fs.default.name"));
        hadoopConfig.set("hadoop.job.ugi", xstorageConfig.getString("hadoop.job.ugi"));
        try {
            FileSystem hadoopFileSystem = FileSystem.get(hadoopConfig);
            BucketDao bucketDao = DaoManager.getInstance().getBucketDao();
            return new HDFSFileSystem(hadoopFileSystem, bucketDao);
        } catch (IOException e) {
            log.error("fail to connect to HDFS", e);
            throw new XStorageRuntimeException("fail to connect to HDFS", e);
        }
    }

    public StorageServlet() {

    }

    @Override
    public void service(HttpServletRequest request, HttpServletResponse servletResponse) throws ServletException,
        IOException {
        if (request.getQueryString() == null) {
            log.debug("{} {}", request.getMethod(), request.getRequestURL());
        } else {
            log.debug("{} {}?{}",
                new Object[] { request.getMethod(), request.getRequestURL(), request.getQueryString() });
        }

        try {
            Result oprationResult = createAndExecuteOperation(request);
            responseResultToClient(oprationResult, servletResponse);
        } catch (XStorageException e) {
            // 重构代码 让XStorageException 把 ErrorResult带到外层
            ErrorResult errorResult = e.getErrorResult();
            if (errorResult == null) {
                log.error("error occured when create and execute operation", e);
                errorResult = ErrorResult.create(ErrorResult.Type.INTERNAL_ERROR);
            }
            responseResultToClient(errorResult, servletResponse);
        }
    }

    /**
     * response the result of operation to client.
     * 
     * @param operationResult
     * @param servletResponse
     * @throws IOException
     */
    private void responseResultToClient(Result operationResult, HttpServletResponse servletResponse) throws IOException {
        // add headers
        for (Entry<String, String> header : operationResult.getHeaders().entries()) {
            servletResponse.addHeader(header.getKey(), header.getValue());
        }
        InputStream input = operationResult.getInputStream();
        OutputStream output = servletResponse.getOutputStream();
        try {
            IOUtils.copy(input, output);
        } catch (IOException e) {
            log.error("error occurred when response the operation result to client", e);
            throw e;
        } finally {
            input.close();
        }
        servletResponse.setStatus(operationResult.getStatus());
        output.flush();
        IOUtils.closeQuietly(output);
    }

    /**
     * create the responding Operation(e.g. PutBucket,GetObject...) and then execute it.
     * 
     * @param request
     * @return the result of operation
     * @throws IOException
     * @throws XStorageException
     */
    private Result createAndExecuteOperation(HttpServletRequest request) throws IOException, XStorageException {
        XBucket bucket = createBucket(request);
        XObject object = createObject(request);

        if (object != null) {
            object.setBucket(bucket);
        }
        log.debug("the requested bucket is [{}]", bucket == null ? "null" : bucket.getName());
        log.debug("the requested object is [{}]", object == null ? "null" : object.getKey());
        ListMultimap<String, String> headers = getHeaders(request);
        Map<String, String> parameters = getParameters(request);

        String method = request.getMethod();
        if (method.equals("PUT") && headers.containsKey("x-amz-copy-source")) {
            method = "COPY";
        }
        // 过期检测
        ExpireChecker.expireCheck(method, headers, parameters);

        // 签名检测
        XUser requester = AuthorizationChecker.check(request.getMethod(), bucket, object, headers, parameters);

        // 判断操作对象(三选一)
        XElement target = determineOperationTarget(service, bucket, object);
        Operation operation = operationFatory.createOperation(method, target, parameters);
        log.debug("create and execute operation[{}]", operation);
        Result result = operation.execute(requester, target, headers, parameters);
        log.debug("the result of operation is [{}], status code is [{}]", result, result.getStatus());
        return result;
    }

    /**
     * 把request转化成响应的XBucket
     * 
     * @param request
     * @return
     * @throws IOException
     */
    private XBucket createBucket(HttpServletRequest request) throws IOException {
        // like : 500doc.s.x-storage.com
        String serverName = request.getServerName().toLowerCase();

        XBucket bucket = new XBucket();
        if (serverName.endsWith(Constants.XS_HOSTNAME_SUFFIX)) {// Virtual Hosted Style
            // remove ".s.x-storage.com"
            String bucketName = serverName.substring(0, serverName.length() - Constants.XS_HOSTNAME_SUFFIX.length());
            bucket.setName(bucketName);
        } else if (serverName.equals(Constants.XS_HOSTNAME)) { // Path Style Method
            String requestURI = request.getRequestURI();
            if (requestURI.length() == 1) { // if == '/'
                return null;
            } else {
                // find the second '/' char
                int secondSlashIndex = requestURI.indexOf(47, 1);
                if (secondSlashIndex == -1) {
                    return null;
                }
                String bucketName = requestURI.substring(1, secondSlashIndex);
                bucket.setName(bucketName);
            }
        } else { // CNAME Method(DNS alias)
            bucket.setName(serverName);
        }

        // Put Bucket acl
        if ("PUT".equals(request.getMethod()) && request.getParameter("acl") != null) {
            String xml = IOUtils.toString(request.getInputStream());
            AccessControlPolicy acp = null;
            try {
                acp = AccessControlPolicy.fromXML(xml);
            } catch (IllegalArgumentException e) {
                // TODO 异常处理
            }
            bucket.setAcp(acp);
        }
        return bucket;
    }

    /**
     * 把request转化成响应的XObject
     * 
     * @param request
     * @return
     * @throws IOException
     * @throws XStorageException
     */
    private XObject createObject(HttpServletRequest request) throws IOException, XStorageException {
        if (request.getMethod().equals("POST")) {
            Map<String, String> postRequestParameterMap = PostObjectRequestParser.parsePostRequestParameters(request);
            String key = postRequestParameterMap.get("key");
            XObject object = new XObject(key);
            InputStream inputStream = PostObjectRequestParser.parsePostRequestInputStream(request);
            object.setInputStream(inputStream);
            return object;
        }

        // like: /2010/01/swf/897897289.swf
        String requestURI = request.getRequestURI();
        String key = null;
        if (requestURI.length() == 1) { // requestURI == '/'
            return null;
        } else if (request.getServerName().equalsIgnoreCase(Constants.XS_HOSTNAME)) {
            // requestURI == /${bucket.name}/${object.key},so find the second '/' char index
            int secondSlashIndex = requestURI.indexOf(47, 1);
            key = requestURI.substring(secondSlashIndex + 1);
        } else {
            key = requestURI.substring(1);
        }

        XObject object = new XObject(key);
        object.setInputStream(request.getInputStream());
        return object;
    }

    private XElement determineOperationTarget(XService service, XBucket bucket, XObject object) {
        if (bucket != null && object != null) {
            return object;
        } else if (bucket != null) {
            // bucket != null && object == null
            return bucket;
        } else {
            // bucket == null && object == null
            return service;
        }
    }

    /**
     * all the key of headers is lowercase, when contains x-amz-date remove date,
     * 
     * @param request
     * @return the headers, all the key of headers is lowercase
     * @throws IOException
     * @throws XStorageException
     */
    @SuppressWarnings("unchecked")
    private ListMultimap<String, String> getHeaders(HttpServletRequest request) throws XStorageException, IOException {
        if ("POST".equals(request.getMethod())) {
            return PostObjectRequestParser.parsePostRequestHeaders(request);
        }
        ListMultimap<String, String> headers = ArrayListMultimap.create();
        Enumeration<String> headerNames = request.getHeaderNames();
        while (headerNames.hasMoreElements()) {
            String headerName = headerNames.nextElement();
            Enumeration<String> headerValues = request.getHeaders(headerName);
            //TODO：这样做是否有错误toLowerCase
            headerName = headerName.toLowerCase();
            while (headerValues.hasMoreElements()) {
                headers.put(headerName, headerValues.nextElement());
            }
        }

        // because the x-amz-date(priority) > date(priority)
        if (headers.containsKey(Constants.REST_METADATA_ALTERNATE_DATE)) {
            headers.removeAll("date");
        }
        return headers;
    }

    /**
     * get the parameters on HTTP request.
     * 
     * @param request
     * @return
     * @throws XStorageException
     * @throws IOException
     */
    @SuppressWarnings("unchecked")
    private Map<String, String> getParameters(HttpServletRequest request) throws XStorageException, IOException {
        if ("POST".equals(request.getMethod())) {
            return PostObjectRequestParser.parsePostRequestParameters(request);
        }
        Map<String, String> parameters = new HashMap<String, String>();
        Enumeration<String> names = request.getParameterNames();
        while(names.hasMoreElements()) {
            String name = names.nextElement();
            parameters.put(name, request.getParameter(name));
        }
        
        return parameters;
    }

}
